package com.gaofans.vertx.boot.dbutils.session.impl;

import com.gaofans.vertx.boot.dbutils.mapper.RowMapper;
import com.gaofans.vertx.boot.dbutils.session.FindOperator;
import io.vertx.sqlclient.*;
import reactor.core.publisher.Mono;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.stream.Collectors;

public class FindOperatorImpl<S extends SqlClient,T> implements FindOperator<S,T> {

    private String sql;
    private RowMapper<T> mapper;
    private List<List<Object>> batchParams;
    private List<Object> params;
    private S sqlClient;

    public FindOperatorImpl(S sqlClient) {
        this.sqlClient = sqlClient;
    }

    @Override
    public FindOperator<S, T> by(String sql) {
        this.sql = sql;
        return this;
    }

    @Override
    public FindOperator<S, T> map(RowMapper<T> mapper) {
        this.mapper = mapper;
        return this;
    }

    @Override
    public FindOperator<S, T> batch(List<Object>... params) {
        this.batchParams = Arrays.asList(params);
        return this;
    }

    @Override
    public FindOperator<S, T> with(Object... params) {
        this.params = Arrays.asList(params);
        return this;
    }

    @Override
    public Mono<T> single() {
        if((params == null || params.size() != 0) && (batchParams == null || batchParams.size() != 0)){
            return Mono.create(sink -> {
                sqlClient.query(sql).execute(res -> {
                    if(res.succeeded()){
                        sink.success(mapper.single(res.result()));
                    }else{
                        sink.error(res.cause());
                    }
                });
            });
        }else {
            return Mono.create(sink -> {
                PreparedQuery<RowSet<Row>> query = sqlClient.preparedQuery(sql);
                if(batchParams != null && batchParams.size() != 0){
                    List<Tuple> tuples = batchParams.stream().map(Tuple::wrap).collect(Collectors.toList());
                    query.executeBatch(tuples,res -> {
                        if(res.succeeded()){
                            sink.success(mapper.single(res.result()));
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }else{
                    sqlClient.preparedQuery(sql).execute(Tuple.wrap(new ArrayList<>(params)),res -> {
                        if(res.succeeded()){
                            sink.success(mapper.single(res.result()));
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }
            });
        }
    }

    @Override
    public Mono<List<T>> list() {
        if((params == null || params.size() != 0) && (batchParams == null || batchParams.size() != 0)){
            return Mono.create(sink -> {
                sqlClient.query(sql).execute(res -> {
                    if(res.succeeded()){
                        sink.success(mapper.list(res.result()));
                    }else{
                        sink.error(res.cause());
                    }
                });
            });
        }else {
            return Mono.create(sink -> {
                PreparedQuery<RowSet<Row>> query = sqlClient.preparedQuery(sql);
                if(batchParams != null && batchParams.size() != 0){
                    List<Tuple> tuples = batchParams.stream().map(Tuple::wrap).collect(Collectors.toList());
                    query.executeBatch(tuples,res -> {
                        if(res.succeeded()){
                            sink.success(mapper.list(res.result()));
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }else{
                    sqlClient.preparedQuery(sql).execute(Tuple.wrap(new ArrayList<>(params)),res -> {
                        if(res.succeeded()){
                            sink.success(mapper.list(res.result()));
                        }else{
                            sink.error(res.cause());
                        }
                    });
                }
            });
        }
    }
}
